package org.sabayframework.cluster.impl;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.InetAddress;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;

import org.sabayframework.cluster.Cluster;
import org.sabayframework.cluster.ClusterContext;
import org.sabayframework.cluster.ClusterEvent;
import org.sabayframework.cluster.ClusterListener;
import org.sabayframework.cluster.ClusteringService;
import org.sabayframework.cluster.Computer;
import org.sabayframework.log.Log;
import org.w3c.dom.Document;
import org.w3c.dom.Element;

public class ClusteringServiceImpl implements ClusteringService, ClusterListener {

	private ClusterState state;
	private LocalComputer computer;
	private int port=34567;
	private int maxBufferSize = 1<<21;
	private int maxQueueSize = 100;
	
	private String clusterName="supercluster";
	private String propertyFile="supercluster.xml";
	
	public ClusteringServiceImpl() {
		super();
	}

	@Override
	public Computer getComputer() {
		return computer;
	}

	@Override
	public Cluster getCluster() {
		return state;
	}
	
	public void stop(){
		computer.stop();
		state.removeEventListener(this);
		ClusterContext.setClusteringService(null);
	}
	
	public void computerJoined(Computer c) {
		Log.info("computerJoined ",c);
	}

	public void computerLeft(Computer c) {
		Log.info("computerLeft ", c);
	}

	public void handleEvent(ClusterEvent event) {
		Log.info("handleEvent ",event);
		if(event.getType() == ClusterEvent.EXIT) {
			stop();
			System.exit(0);
		}
	}
	
	public int getPort() {
		return port;
	}
	public void setPort(int port) {
		this.port = port;
	}
	
	public int getMaxBufferSize() {
		return maxBufferSize;
	}

	public void setMaxBufferSize(int maxBufferSize) {
		this.maxBufferSize = maxBufferSize;
	}

	public int getMaxQueueSize() {
		return maxQueueSize;
	}

	public void setMaxQueueSize(int maxQueueSize) {
		this.maxQueueSize = maxQueueSize;
	}

	public String getClusterName() {
		return clusterName;
	}
	public void setClusterName(String clusterName) {
		this.clusterName = clusterName;
	}
	
	public String getPropertyFile() {
		return propertyFile;
	}
	public void setPropertyFile(String propertyFile) {
		this.propertyFile = propertyFile;
	}
	public void start() throws Exception{
		File file = new File(propertyFile);
		InputStream is =  null;
		if(!file.canRead()){
			is =  getClass().getClassLoader().getResourceAsStream(propertyFile);
		}
		if(is == null) is = new FileInputStream(file); 
		
		DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
		DocumentBuilder builder = factory.newDocumentBuilder();
		Document doc=builder.parse(is);
		Element element=doc.getDocumentElement();
		
		
		state = new ClusterState(clusterName, element);

		ComputerIdImpl id = new ComputerIdImpl(InetAddress.getLocalHost(), port);
		if(state.hasChild(id)){
			throw new IllegalArgumentException("Local computer on the same port ("+id.getPort()+") has already joined the cluster. Please choose different port.");
		}
		computer = new LocalComputer(id);
		
		state.addEventListener(this);
		ClusterContext.setClusteringService(this);
		
		computer.start();
		state.startExecutionQueue(computer, maxBufferSize, maxQueueSize);
			
	}
	

}
